package com.atguigu.tingshu.album.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.AlbumStatVoMsg;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;

@CanalEventListener  //声明当前类为canal的一个事件监听器 可以监听数据库的增删改操作，但不包括读操作
@Slf4j
public class CanalClientListener {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    KafkaService kafkaService;

    //监听album_info数据表中的增删改变动，并将监听的结果封装rowData用于从redis中删除保存的数据
    @ListenPoint(destination = "example",schema = "tingshu_album",table = "album_info")
    public void albumInfoUpdateListener(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        /**
         * rowData 中封装了修改之前和之后的数据
         */
//        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
//        beforeColumnsList.forEach(column -> {
//            String columnName = column.getName();
//            String columnValue = column.getValue();
//            log.info("columnName:{},columnValue:{}",columnName,columnValue);
//        });
//        log.info("----------------------------------------------------------------");
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            if ("id".equals(column.getName())) {
                stringRedisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX+column.getValue());
            }
            log.info("使用canal中从redis中删除了专辑的基本信息");
        });
    }

    //监听album_stat数据表中的增删改变动，并将监听的结果封装rowData用于从redis中删除保存的数据
    @ListenPoint(destination = "example",schema = "tingshu_album",table = "album_stat")
    public void albumStatUpdateListener(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

        //同步ES中专辑的统计数据
        //使用kafka发送消息，通过消费者search微服务同步ES中的专辑统计信息
        AlbumStatVoMsg albumStatVoMsg = new AlbumStatVoMsg();

        afterColumnsList.forEach(column -> {
            if ("album_id".equals(column.getName())) {
                stringRedisTemplate.delete(RedisConstant.ALBUM_STAT_PREFIX+column.getValue());
                albumStatVoMsg.setAlbumId(Long.parseLong(column.getValue()));
            }
            log.info("使用canal中从redis中删除了专辑的统计信息");

            //遍历表的列字段获取kafka发送消息端的属性值
            if ("stat_type".equals(column.getName())){
                albumStatVoMsg.setStatType(column.getValue());
            }
            if ("stat_num".equals(column.getName())){
                albumStatVoMsg.setStatNum(Integer.parseInt(column.getValue()));
            }
            log.info("使用canal遍历表的列字段获取kafka发送消息端的属性值");
        });
        kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_ES_STAT_UPDATE_SYNC, JSON.toJSONString(albumStatVoMsg));
        log.info("使用canal通过kafka消息队列修改ES中的专辑统计信息数据...{}",JSON.toJSONString(albumStatVoMsg));
    }

    //删除redis中的专辑分类数据信息
    @ListenPoint(destination = "example",schema = "tingshu_album",table = "base_category_view")
    public void categoryViewUpdateListener(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            if ("id".equals(column.getName())) {
                stringRedisTemplate.delete(RedisConstant.CATEGORY_INFO_PREFIX+column.getValue());
            }
            log.info("使用canal中从redis中删除了专辑分类数据信息");
        });
    }

    //删除redis中用户基本信息
    @ListenPoint(destination = "example",schema = "tingshu_user",table = "user_info")
    public void UserInfoUpdateListener(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            if ("id".equals(column.getName())) {
                stringRedisTemplate.delete(RedisConstant.USER_INFO_PREFIX+column.getValue());
            }
            log.info("使用canal中从redis中删除了用户基本数据信息");
        });
    }



}
